package com.yanggu.flink.datastream_api.sink

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.rabbitmq.RMQSink
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig

/**
 * 把flink的DataStream数据输出到rabbitmq中
 */
object RabbitMQSinkDemo {

  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment

    val tool = ParameterTool.fromPropertiesFile(getClass.getResourceAsStream("/rabbitmq.properties"))

    val config = new RMQConnectionConfig.Builder()
      .setHost(tool.get("rabbitmq.host"))
      .setPort(tool.getInt("rabbitmq.port"))
      .setVirtualHost(tool.get("rabbitmq.virtualHost"))
      .setUserName(tool.get("rabbitmq.username"))
      .setPassword(tool.get("rabbitmq.password"))
      .build()

    val rabbitmqSink = new RMQSink[String](config, tool.get("rabbitmq.queue"), new SimpleStringSchema)

    environment
      .fromCollection(List("{123}"))
      .addSink(rabbitmqSink)

    environment.execute("RabbitMQSinkDemo Job")

  }

}
